package com.zyfu.demo;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.messaging.MessageHeaders;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;

/**
 * <b>请输入名称</b>
 * <pre>
 * 描述<br/>
 * 作用：；<br/>
 * 限制：；<br/>
 * </pre>
 *
 * @author 张永福
 * @since  2024/3/14 09:50
 */
@SpringBootApplication
@EnableKafka
public class KafkaApplication {
    public static void main(String[] args) {
        SpringApplication.run(KafkaApplication.class, args);
    }
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Bean
    public CustomProducer producer() {
        CustomProducer producer = new CustomProducer();
        producer.setKafkaTemplate(kafkaTemplate);
        producer.sendMsg("zyfu.test", "hello zyfu !");
        return producer;
    }

    @KafkaListener(topics = "zyfu.test", groupId = "zyfu_test_group", id = "zyfuConsumer")
    public void reciveHello(MessageHeaders messageHeaders, String data) {
        System.out.println(messageHeaders.getId());
        Long timestamp = messageHeaders.get("kafka_receivedTimestamp", Long.class);
        LocalDateTime localDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.of("Asia/Shanghai"));
        System.out.println(localDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd hh:mm:ss")));
        System.out.println("接收到消息:" + data);
    }
}
